﻿#include"Client.hpp"
#include"./Malloc/TypesDef.hpp"
#include<iostream>


namespace MMFNetwork
{

    Client::Client(
            const std::string& remoteIp,
            uint16_t remotePort,
            uint32_t bufferSize,
            ClientErrorHandler errHandler,
            ClientStatusHandler statusHandler,
            ClientAsyncHandler asyncHandler,
            ClientPackageHandler pkgHandler,
            MMFInetFamily family,
            bool Sym
        )
        : mService(1)       //客户端只需要一个并发
        , mRemoteIp(remoteIp)
        , mBufferSize(bufferSize)
        , mCurrentRecvBytes(0)
        , mCurrentBuff(0)           ///< 当前使用的缓冲
        , mRecvBuffers(128)           ///< 接收缓冲数
        , mRemotePort(remotePort)
        , mErrHandler(errHandler)
        , mStatusHandler(statusHandler)
        , mAsyncHandler(asyncHandler)
        , mPkgHandler(pkgHandler)
        , mFamily(family)
        , mSocket(mService)
        , mStatus(Status::Invalid)
        , mIsStop(false)
        , mIsSym(Sym)
    {
        for (uint32_t i = 0; i < mRecvBuffers.size(); i++)
            mRecvBuffers[i] = aligned_alloc(PAGE_SIZE, bufferSize);
    }

    Client::~Client()
    {
        Stop();
        for (uint32_t i = 0; i < mRecvBuffers.size(); i++)
            aligned_free(mRecvBuffers[i]);
    }

    bool Client::Start()
    {
        try
        {
            //创建链接的端信息
            auto endpoint = ip::tcp::endpoint(mFamily==MMFInetFamily::V4? ip::tcp::v4() : ip::tcp::v6(), mRemotePort);
            endpoint.address(ip::address::from_string(mRemoteIp));      //设置ip
            mSocket.connect(endpoint);
        }
        catch(boost::system::system_error err)
        {
            //链接出错
            boost::system::error_code ec = err.code();
            mErrHandler(ec);
            mStatus = MMFNetwork::Status::Invalid;
            mStatusHandler(mStatus);
            return false;
        }

        //链接成功
        mStatus = MMFNetwork::Status::Connected;
        mStatusHandler(mStatus);

        boost::system::error_code ec;
        mLocalIp = mSocket.local_endpoint(ec).address().to_string();
        mRemotePort = mSocket.remote_endpoint(ec).port();

        //开始接收数据
        if (!mIsSym)
            AsyncRecvPackage();

        //创建子线程运行asio_ctx
        mWorker = std::make_shared<std::thread>([this](){
            //运行服务，这里时各循环会一直运行
            while(!mIsStop)
            {
                mService.run();
                std::this_thread::sleep_for(std::chrono::milliseconds{10});
            }
        });

        return true;
    }

    void Client::Stop()
    {
        mStatus = MMFNetwork::Status::Invalid;
        mIsStop = true;
        std::this_thread::sleep_for(std::chrono::milliseconds{10});

        //断开套接字
        mSocket.close();

        //停止asio服务
        mService.stop();

        //停止线程
        if (mWorker && mWorker->joinable())
        {
            mWorker->join();
            mWorker.reset();
        }
    }

    const std::string& Client::LocalIp()
    {
        return mLocalIp;
    }

    const std::string& Client::RemoteIp()
    {
        return mRemoteIp;
    }

    uint16_t Client::LocalPort()
    {
        return mLocalPort;
    }

    uint16_t Client::RemotePort()
    {
        return mRemotePort;
    }

    Status Client::CurrentStatus()
    {
        return mStatus;
    }

    bool Client::Send(void* buff, uint32_t count)
    {
        boost::system::error_code ec;
        boost::asio::write(mSocket, boost::asio::buffer(buff, count), ec);
        if (ec)
        {   
            mStatus = MMFNetwork::Status::WriteErr;
            mStatusHandler(mStatus);
            return false;
        }

        return true;
    }

    bool Client::Recv(void* buff, uint32_t count)
    {
        boost::system::error_code ec;
        boost::asio::read(mSocket, boost::asio::buffer(buff, count), ec);
        if (ec)
        {   
            mStatus = MMFNetwork::Status::ReadErr;
            mStatusHandler(mStatus);
            return false;
        }

        return true;
    }

    void Client::AsyncSend(void* buff, uint32_t count)
    {
        boost::asio::async_write(mSocket, boost::asio::buffer(buff, count), 
        [this, count, buff](boost::system::error_code ec, size_t sentNum){
            if (ec || sentNum < count)  //发送出错或者发送字节数与数据字节数不一致
            {
                mStatus = MMFNetwork::Status::AsyncWriteErr;
                mStatusHandler(mStatus);
            }
        });
    }

    void Client::AsyncRecv(void* buff, uint32_t count, const ClientAsyncHandler& handler)
    {
        boost::asio::async_read(mSocket, boost::asio::buffer(buff, count),
        [this, count, buff, handler](boost::system::error_code ec, size_t sentNum){
            if (ec || sentNum < count)  //接收出错或者接收字节数与数据字节数不一致
            {
                mStatus = MMFNetwork::Status::AsyncReadErr;
                mStatusHandler(mStatus);
            }
            else
                handler(buff, true);
        });
    }

    void Client::AsyncRecvPackage()
    {
        //假定需要收取这么大
        size_t recvBytes = mBufferSize - mCurrentRecvBytes;
        void* recvBuff = mRecvBuffers[mCurrentBuff];

        mSocket.async_receive(
            boost::asio::buffer((char*)recvBuff + mCurrentRecvBytes, recvBytes), //接着收数据
            [this, recvBuff](boost::system::error_code ec, size_t num){

                if (ec)  //接收出错
                {
                    mStatus = MMFNetwork::Status::AsyncReadErr;
                    mStatusHandler(mStatus);
                    return;
                }

                //增加接收数量
                mCurrentRecvBytes += (uint32_t)num;

                //判断是否收满了头
                if (mCurrentRecvBytes > sizeof(MMFMessageHeader))
                {
                    //转换指针类型
                    MMFMessageHeader* head = (MMFMessageHeader*)recvBuff;

                    //判断数据是否收满
                    uint32_t dataLen = head->mDataBytes;

                    //如果数据已经收满
                    if (mCurrentRecvBytes >= dataLen)
                    {
                        //处理多个消息被收满的情况
                        uint32_t next = 0;
                        do 
                        {
                            //已经收的数据是一个完整的包
                            mCurrentRecvBytes = mCurrentRecvBytes - dataLen;

                            //通知包数据已经收完
                            mPkgHandler((char*)recvBuff + next, dataLen);
                            next += dataLen;

                            //如果不够头长度说明，没有消息
                            if (mCurrentRecvBytes < sizeof(MMFMessageHeader))
                                break;
                            
                            //计算下一个数据的位置
                            head = (MMFMessageHeader*)((char*)recvBuff + next);
                            dataLen = head->mDataBytes;

                        }while (mCurrentRecvBytes >= dataLen);

                        //交换接收缓冲缓冲
                        mCurrentBuff++;
                        mCurrentBuff %= mRecvBuffers.size();
                        void* newBuff = mRecvBuffers[mCurrentBuff];

                        //拷贝下一个数据
                        if (mCurrentRecvBytes > 0)
                            memcpy(newBuff, (char*)recvBuff + dataLen, mCurrentRecvBytes);
                    }
                }

                AsyncRecvPackage();
            }
        );
    }
}